package ma.zhiqiang.dataline.connector.sink;

import cn.hutool.core.util.ClassUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import ma.zhiqiang.dataline.core.dto.ColumnMap;
import ma.zhiqiang.dataline.core.dto.SinkRecord;
import ma.zhiqiang.dataline.core.engine.Sink;
import ma.zhiqiang.dataline.core.format.WriterFormatCsv;
import ma.zhiqiang.dataline.core.format.WriterFormatJson;
import org.apache.commons.io.FileUtils;

import java.io.File;
import java.io.IOException;
import java.util.Properties;

@Slf4j
public class FileSink implements Sink {
    Properties properties;
    String path;
    String format;
    String encoding;
    File output;

    public FileSink(Properties properties) {
        this.properties = properties;
        this.path = properties.getProperty("path");
        this.format = properties.getProperty("format","text");
        this.encoding = properties.getProperty("encoding","UTF-8");
        this.output = new File(path);
    }

    @Override
    public Boolean writer(SinkRecord sinkRecord) {
        String outline = "";
        if("csv".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatCsv().format(sinkRecord);
        }else if("json".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatJson().format(sinkRecord);
        }else if(sinkRecord.getData() instanceof String){  //"none".equals(format)
            outline = sinkRecord.getData().toString();
        }else if(sinkRecord.getData() instanceof ColumnMap){
            outline = new WriterFormatJson().format(sinkRecord);
        }
        log.info(outline);
        try {
            FileUtils.write(output,outline+"\n",encoding,true);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return Boolean.TRUE;
    }
}
